#include "thread_pool.h"
#define DEBUG
void handler(void *arg)
{
	printf("[%u] is ended.\n",
				 (unsigned)pthread_self());

	pthread_mutex_unlock((pthread_mutex_t *)arg);
}

void *routine(void *arg)
{
#ifdef DEBUG
	printf("[%u] is started.\n",
				 (unsigned)pthread_self());
#endif
	// 获取线程池结构体
	thread_pool *pool = (thread_pool *)arg;
	struct task *p; // 野指针 后续指向任务

	while (1)
	{
		/*
		** push a cleanup functon handler(), make sure that
		** the calling thread will release the mutex properly
		** even if it is cancelled during holding the mutex.
		**
		** NOTE:
		** pthread_cleanup_push() is a macro which includes a
		** loop in it, so if the specified field of codes that
		** paired within pthread_cleanup_push() and pthread_
		** cleanup_pop() use 'break' may NOT break out of the
		** truely loop but break out of these two macros.
		** see line 61 below.
		*/
		//================================================//
		pthread_cleanup_push(handler, (void *)&pool->lock); // 注册取消处理函数 防止死锁
		pthread_mutex_lock(&pool->lock);
		//================================================//

		// 1, no task, and is NOT shutting down, then wait
		while (pool->waiting_tasks == 0 && !pool->shutdown)
		{
			pthread_cond_wait(&pool->cond, &pool->lock); // 休眠 自动解锁
		}

		// 2, no task, and is shutting down, then exit 没任务就关机或退出
		if (pool->waiting_tasks == 0 && pool->shutdown == true)
		{
			pthread_mutex_unlock(&pool->lock);
			pthread_exit(NULL); // CANNOT use 'break';
		}

		// 3, have some task, then consume it
		p = pool->task_list->next;
		pool->task_list->next = p->next;
		pool->waiting_tasks--;

		//================================================//
		pthread_mutex_unlock(&pool->lock); // 防止线程抢夺任务
		pthread_cleanup_pop(0);						 // 移除取消处理函数
		//================================================//

		pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL); // 禁止线程被取消
		(p->do_task)(p->arg);																	// 执行任务
		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);	// 允许线程被取消

		free(p); // 释放任务节点
	}

	pthread_exit(NULL);
}

bool init_pool(thread_pool *pool, unsigned int threads_number)
{
	pthread_mutex_init(&pool->lock, NULL);
	pthread_cond_init(&pool->cond, NULL);

	pool->shutdown = false;																			 // 关机标志
	pool->task_list = malloc(sizeof(struct task));							 // 分配任务链表头
	pool->tids = malloc(sizeof(pthread_t) * MAX_ACTIVE_THREADS); // 分配线程id数组

	if (pool->task_list == NULL || pool->tids == NULL)
	{
		perror("allocate memory error");
		return false;
	}

	pool->task_list->next = NULL; // 任务链表初始化

	pool->max_waiting_tasks = MAX_WAITING_TASKS; //  初始化最大等待任务数
	pool->waiting_tasks = 0;										 // 当前等待任务数
	pool->active_threads = threads_number;			 // 当前活动线程数

	int i;
	for (i = 0; i < pool->active_threads; i++) // 循环创建线程
	{
		if (pthread_create(&((pool->tids)[i]), NULL,
											 routine, (void *)pool) != 0)
		{
			perror("create threads error");
			return false;
		}

#ifdef DEBUG
		printf("[%u]:[%s] ==> tids[%d]: [%u] is created 线程已创建.\n",
					 (unsigned)pthread_self(), __FUNCTION__,
					 i, (unsigned)pool->tids[i]);
#endif
	}

	return true;
}

bool add_task(thread_pool *pool,
							void *(*do_task)(void *arg), void *arg)
{
	// 新建任务节点
	struct task *new_task = malloc(sizeof(struct task));
	if (new_task == NULL)
	{
		perror("allocate memory error");
		return false;
	}
	new_task->do_task = do_task;
	new_task->arg = arg;
	new_task->next = NULL;

	//============ LOCK =============//
	pthread_mutex_lock(&pool->lock); // 加锁 防止添加任务时其他线程取任务/线程池销毁
	//===============================//
	// 超出最大任务数进 放弃任务
	if (pool->waiting_tasks >= MAX_WAITING_TASKS)
	{
		pthread_mutex_unlock(&pool->lock);

		fprintf(stderr, "too many tasks.\n");
		free(new_task);

		return false;
	}
	// 指向任务链表头
	struct task *tmp = pool->task_list;
	while (tmp->next != NULL) // 遍历找到链尾
		tmp = tmp->next;

	tmp->next = new_task;
	pool->waiting_tasks++;

	//=========== UNLOCK ============//
	pthread_mutex_unlock(&pool->lock);
	//===============================//

#ifdef DEBUG
	printf("[%u][%s] ==> 一个新任务已加入.\n",
				 (unsigned)pthread_self(), __FUNCTION__);
#endif

	pthread_cond_signal(&pool->cond);
	return true;
}

int add_thread(thread_pool *pool, unsigned additional_threads)
{
	if (additional_threads == 0)
		return 0;

	unsigned total_threads =
			pool->active_threads + additional_threads;

	int i, actual_increment = 0;
	// 遍历线程池，找到空位
	for (i = pool->active_threads;
			 i < total_threads && i < MAX_ACTIVE_THREADS;
			 i++)
	{
		if (pthread_create(&((pool->tids)[i]),
											 NULL, routine, (void *)pool) != 0)
		{
			perror("add threads error");

			// no threads has been created, return fail
			if (actual_increment == 0)
				return -1;

			break;
		}
		actual_increment++;

#ifdef DEBUG
		printf("[%u]:[%s] ==> tids[%d]: [%u] is created.\n",
					 (unsigned)pthread_self(), __FUNCTION__,
					 i, (unsigned)pool->tids[i]);
#endif
	}

	pool->active_threads += actual_increment;
	return actual_increment;
}

int remove_thread(thread_pool *pool, unsigned int removing_threads)
{
	if (removing_threads == 0) // 线程池线程
		return pool->active_threads;

	int remaining_threads = pool->active_threads - removing_threads;	 // 剩余线程数
	remaining_threads = remaining_threads > 0 ? remaining_threads : 1; // 至少保留一个线程

	int i;
	for (i = pool->active_threads - 1; i > remaining_threads - 1; i--) // 从后往前删除线程
	{
		errno = pthread_cancel(pool->tids[i]);

		if (errno != 0)
			break;
		// 打印取消线程失败
#ifdef DEBUG
		printf("[%u]:[%s] ==> cancelling tids[%d]: [%u]...\n",
					 (unsigned)pthread_self(), __FUNCTION__,
					 i, (unsigned)pool->tids[i]);
#endif
	}

	if (i == pool->active_threads - 1) // 删除线程失败
		return -1;
	else
	{
		pool->active_threads = i + 1; // 更新线程池线程数
		return i + 1;
	}
}

bool destroy_pool(thread_pool *pool)
{
	// 1, activate all threads
	pool->shutdown = true;
	pthread_cond_broadcast(&pool->cond); // 唤醒所有线程

	// 2, wait for their exiting
	int i;
	for (i = 0; i < pool->active_threads; i++)
	{
		errno = pthread_join(pool->tids[i], NULL); // 等待线程结束
		if (errno != 0)
		{
			printf("join tids[%d] error: %s\n",
						 i, strerror(errno));
		}
		else
			printf("[%u] is joined\n", (unsigned)pool->tids[i]); // 打印线程结束
	}

	// 3, free memories //释放内存
	free(pool->task_list);
	free(pool->tids);
	free(pool);

	return true;
}
